// Author: unmerged@gmail.com (Hui Chen)
//
// LevelDB server.

#include <boost/algorithm/string.hpp>
#include <boost/format.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/unordered_map.hpp>
#include <gflags/gflags.h>
#include <glog/logging.h>
#include <leveldb/db.h>
#include <leveldb/iterator.h>
#include <server1/server/client_connection.hpp>
#include <server1/server/server.hpp>
#include <server1/server/protobuf_connection.hpp>

#include <string>

#include "leveldb_service.pb.h"

DEFINE_string(db_name,
              "test_db",
              "LevelDB database name.");

DEFINE_string(address,
              "localhost",
              "Address of the server.");

DEFINE_string(port,
              "9999",
              "Port that the server will listen to.");

DEFINE_int32(num_threads,
             4,
             "Number of RPC server threads.");

using namespace boost;

namespace levelserv {

class LevelDbServiceImpl : public LevelDbService {
 public:
  LevelDbServiceImpl() {
    // Open database.
    leveldb::Options options;
    leveldb::Status status;
    options.create_if_missing = true;
    status = leveldb::DB::Open(options, FLAGS_db_name, &db);
    CHECK(status.ok()) << "Can't open database.";
  }

  ~LevelDbServiceImpl() {
    delete db;
  }

  virtual void Get(
      google::protobuf::RpcController *controller,
      const GetRequest* request,
      GetResponse* response,
      google::protobuf::Closure* done) {
    for (int i = 0; i < request->key_size(); ++i) {
      string value;
      if (request->key(i).empty()) {
        value = "";
      } else {
        string key = str(format("%s:%s")
                         % request->key_namespace()
                         % request->key(i));
        db->Get(leveldb::ReadOptions(), leveldb::Slice(key), &value);
      }
      levelserv::KeyValue* kv = response->add_kv();
      kv->set_key(request->key(i));
      kv->set_value(value);
    }
    done->Run();
  }

  virtual void Set(
      google::protobuf::RpcController *controller,
      const SetRequest* request,
      SetResponse* response,
      google::protobuf::Closure* done) {
    for (int i = 0; i < request->kv_size(); ++i) {
      if (request->kv(i).key().empty()) {
        continue;
      }
      string key = str(format("%s:%s")
                       % request->key_namespace()
                       % request->kv(i).key());
      db->Put(leveldb::WriteOptions(),
              leveldb::Slice(key),
              request->kv(i).value());
    }

    response->set_successful(true);
    done->Run();
  }

 private:
  leveldb::DB* db;
};

}  // namespace levelserv

int main(int argc, char **argv) {
  google::ParseCommandLineFlags(&argc, &argv, true);
  google::InitGoogleLogging(argv[0]);

  CHECK(!FLAGS_db_name.empty()) << "--db_name can't be empty.";

  // RPC service.
  boost::shared_ptr<ProtobufConnection> server_connection;
  server_connection.reset(new ProtobufConnection("Server"));
  levelserv::LevelDbServiceImpl leveldb_service;
  server_connection->RegisterService(&leveldb_service);
  boost::shared_ptr<Server> server;
  server.reset(new Server(1, FLAGS_num_threads));
  server->Listen(FLAGS_address, FLAGS_port, server_connection.get());

  // Server loop.
  sigset_t new_mask;
  sigfillset(&new_mask);
  sigset_t old_mask;
  sigset_t wait_mask;
  sigemptyset(&wait_mask);
  sigaddset(&wait_mask, SIGINT);
  sigaddset(&wait_mask, SIGQUIT);
  sigaddset(&wait_mask, SIGTERM);
  pthread_sigmask(SIG_BLOCK, &new_mask, &old_mask);
  pthread_sigmask(SIG_SETMASK, &old_mask, 0);
  pthread_sigmask(SIG_BLOCK, &wait_mask, 0);
  int sig = 0;
  sigwait(&wait_mask, &sig);
}
